/*
 *  Copyright 2020-2025 the original author or authors.
 *  You cannot use this file unless authorized by the author.
 */

package org.ipig.computing.spark.cluster.yarn;

import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.ipig.commons.service.ProcessorService;
import org.ipig.computing.spark.GenericSparkLauncher;
import org.ipig.computing.spark.conf.ApplicationConf;
import org.ipig.constants.ResultStatus;
import org.ipig.constants.SymbolCnst;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * GenericSparkOnYarnKafkaProcessor
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @since 1.0
 */
@Slf4j
public class GenericSparkOnYarnKafkaProcessor implements ProcessorService<ConsumerRecords<String, String>, ResultStatus> {
    private static final int MAXINUM_POOL_SIZE = Runtime.getRuntime().availableProcessors();
    private static int maxinumPoolSize =MAXINUM_POOL_SIZE;
    private static BlockingQueue queue = new ArrayBlockingQueue(MAXINUM_POOL_SIZE);
    private static volatile ExecutorService executorService = null;
    @Setter
    private GenericSparkLauncher launcher;

    @Override
    public ResultStatus process(ConsumerRecords<String, String> records) {
        if (executorService == null) {
            if (maxinumPoolSize <4) {
                maxinumPoolSize =4;
            }
            int corePoolSize = maxinumPoolSize/2;
            executorService = new ThreadPoolExecutor(4, maxinumPoolSize, 0, TimeUnit.SECONDS, queue);
        }
        if (records != null && !records.isEmpty()) {
            int total = records.count();
            Map<String, ApplicationConf> distinctMap = new HashMap<String, ApplicationConf>();
            records.forEach(t -> {
                String value = t.value();
                try {
                    if (StringUtils.isNotBlank(value)) {
                        ApplicationConf conf = ApplicationConf.toObject(value);
                        distinctMap.put(conf.getMainClass() + SymbolCnst.AT + conf.getAppName(), conf);
                    } else {
                        log.warn("Current data content is empty and not processed");
                    }
                } catch (Exception ex) {
                    log.error("Current data content:{} ,Error converting to ApplicationConf, exception content:{}", value, ex.getMessage());
                }
            });
            if (!distinctMap.isEmpty()) {
                int count = distinctMap.size();
                List<Future<ResultStatus>> list = new ArrayList<>(count);
                for (ApplicationConf v : distinctMap.values()) {
                    int depthQueue = 0;
                    while ((depthQueue = queue.size()) >= maxinumPoolSize)
                        try {
                            log.warn("the current thread pool's blocking queue depth is: " + depthQueue);
                            Thread.sleep(1000);
                        } catch (InterruptedException ex) {
                            ex.printStackTrace();
                        }
                    //如果有异常直接抛出到上一层进行处理
                    Future<ResultStatus> future = executorService.submit(()->{return launcher.launch(v);});
                    list.add(future);
                }
                for (int i = 0; i < count; i++) {
                    try {
                        list.get(i).get();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (ExecutionException e) {
                        throw new IllegalArgumentException("必须处理，否则可能会阻塞消息处理，错误消息：" + e.getMessage());
                    }
                }
                log.info("current message total qty={},去重后成功处理的qty={}", total, count);
            }
        }
        return ResultStatus.SUCCESS;
    }
}

